﻿using Confluent.Kafka;
using System;
using System.Collections.Generic;
using System.Threading;

namespace CommonUtils
{
    /*
     * 启动kafka-server
     * 配置java_home
     * 注意程序目录不能包含任何特殊字符，尽量放在磁盘根目录
     * https://blog.csdn.net/woshixiazaizhe/article/details/80610432
     */

    public static class MqKafka
    {
        public static void Test()
        {
            var config = new Dictionary<string, string>();
            config.Add("bootstrap.servers", "127.0.0.1:9092");
            config.Add("group.id", "DrewGroup");

            //消费者
            var consumerBuilder = new ConsumerBuilder<string, string>(config);
            var consumer = consumerBuilder.Build();
            consumer.Subscribe("aaa");
            new Thread(delegate ()
            {
                while (true)
                {
                    var result = consumer.Consume();
                    Console.WriteLine("消费");
                    Console.WriteLine(result.Message.Key + ": " + result.Message.Value);
                }
            }).Start();

            //生产者
            var producerBuilder = new ProducerBuilder<string, string>(config);
            var producer = producerBuilder.Build();

            while (true)
            {
                Console.WriteLine("任意键生产");
                Console.ReadKey();
                producer.Produce("aaa", new Message<string, string>()
                {
                    Key = "aaa",
                    Value = "aaa"
                });
            }
        }
    }
}
